package org.ala.tiktools.rpc.kcp.netty.client;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.spi.SelectorProvider;
import java.util.ArrayList;
import java.util.List;

import org.ala.tiktools.rpc.kcp.config.RpcKcpConfigInfo;
import org.ala.tiktools.rpc.kcp.netty.tools.CodecOutputList;
import org.ala.tiktools.rpc.kcp.netty.tools.NettyTools;
import org.ala.tiktools.tools.Assert;
import org.ala.tiktools.tools.TimestampTools;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.ChannelConfig;
import io.netty.channel.ChannelException;
import io.netty.channel.ChannelMetadata;
import io.netty.channel.ChannelOutboundBuffer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.RecvByteBufAllocator;
import io.netty.channel.nio.AbstractNioMessageChannel;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.internal.PlatformDependent;
import io.netty.util.internal.SocketUtils;

/**
 * 负责与udp层打交道
 * <p>  1）将字节走udp口send出去<br/>
 *          被KcpClientOutput触发，里面会调用udpChannel.unsafe.write和flush<br/>
 *          flush会走到这里的onWriteMessage<br/>
 * <p>  2）从udp口读字节出来<br/>
 *          被系统触发，入口在unsafe.read()<br/>
 *          unsafe.read()中会执行：<br/>
 *              1）从udp口读出字节块<br/>
 *              2）调用input()，把字节快存进rcvBuf<br/>
 *              3）调用receive()<br/>
 *                  - 根据当前rcvWnd窗口容量，把rcvQue中存储的字节块读出来，并触发fireChannelRead<br/>
 *                  - 读走了多少，从rcvBuf中补多少进rcvQue
 * <p>  3）不检测唯一性。如果jvm内存在多个channel，udp的数据包会广播给所有UdpChannel
 *
 * @author ala
 * @date 2025-04-15 22:46
 */
public class KcpClientUdpChannel extends AbstractNioMessageChannel {
    /**
     *
     */
    private static final ChannelMetadata METADATA = new ChannelMetadata(false);
    /**
     *
     */
    private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();
    /**
     *
     */
    protected boolean inputShutdown;
    /**
     *
     */
    protected KcpClientChannel kcpChannel;


    /**
     *  构建 udpChannel
     */
    private static DatagramChannel newSocket(SelectorProvider provider) {
        try {
            return provider.openDatagramChannel();
        } catch (IOException e) {
            throw new ChannelException("无法打开channel", e);
        }
    }
    public KcpClientUdpChannel(KcpClientChannel ukcpChannel) {
        this(ukcpChannel, newSocket(DEFAULT_SELECTOR_PROVIDER));
    }
    public KcpClientUdpChannel(KcpClientChannel ukcpChannel, SelectorProvider provider) {
        this(ukcpChannel, newSocket(provider));
    }
    public KcpClientUdpChannel(KcpClientChannel ukcpChannel, DatagramChannel udpChannel) {
        super(null, udpChannel, SelectionKey.OP_READ);
        this.kcpChannel = ukcpChannel;
    }


    @Override
    protected void doBeginRead() throws Exception { if (inputShutdown) { return; } else {super.doBeginRead();} }
    /**
     *  从udp socket中读取ByteBuf，读到bufs[i]中
     */
    @SuppressWarnings("deprecation")
	@Override
    protected int doReadMessages(List<Object> bufs) throws Exception {
        DatagramChannel dc = javaChannel();
        ChannelConfig config = config();
        RecvByteBufAllocator.Handle allocatorHandle = unsafe().recvBufAllocHandle();
        //  预分配一块定长内存
        ByteBuf buf = allocatorHandle.allocate(config.getAllocator());
        allocatorHandle.attemptedBytesRead(buf.writableBytes());
        boolean free = true;
        try {
            //  创建写区视图，数据将写在写区视图中
            ByteBuffer buffer = buf.internalNioBuffer(buf.writerIndex(), buf.writableBytes());
            int pos = buffer.position();
            int len = dc.read(buffer);
            if (len <= 0) {
                return len;
            }
            len = buffer.position() - pos;
            allocatorHandle.lastBytesRead(len);
            //  重置buf的写针（右移到本次读取的数据量，的下一个字节位置。右移部分就是ByteBuf的可读区）
            bufs.add(buf.writerIndex(buf.writerIndex() + allocatorHandle.lastBytesRead()));
            free = false;
            return len;
        } catch (Throwable cause) {
            PlatformDependent.throwException(cause);
            return -1;
        } finally {
            if (free) {
                buf.release();
            }
        }
    }

    /**
     * 将数据通过udp口send出去
     * <p>  KcpClientOutput里，会调用udpChannel.write()，并且flush()一次
     * <p>  flush()会具体走到这里，这里就是走udp口send数据
     */
    @Override
    protected boolean doWriteMessage(Object msg, ChannelOutboundBuffer in) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        //  如果没有字节，直接成功
        int len = buf.readableBytes();
        if (len == 0) { return true; }
        //  javaChannel返回的是UdpChannel，走udp口write出去
        ByteBuffer buffer = buf.internalNioBuffer(buf.readerIndex(), len);
        len = javaChannel().write(buffer);
//        log.info("[KcpClientUdpChannel doWriteMessage] 客户端发送数据:({})字节", len);
        return len > 0;
    }

    @Override
    protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
        if (localAddress != null) {
            doBind(localAddress);
        }
        boolean success = false;
        try {
            //  逻辑上的链接，只保持remoteAddress，毕竟udp（只接受来自remoteAddress的消息）
            javaChannel().connect(remoteAddress);
            success = true;
            //  设置update时间戳
            int current = TimestampTools.milliSeconds32();
            int tsUp = kcpChannel.kcpCheck(current);
            kcpChannel.kcpTsUpdate(tsUp);
            kcpChannel.scheduleUpdate(tsUp, current);
            return true;
        } finally {
            if (!success) {
                close();
            }
        }
    }
    @Override
    protected Object filterOutboundMessage(Object msg) {
        if (msg instanceof ByteBuf) {
            ByteBuf buf = (ByteBuf) msg;
            if (isSingleDirectBuffer(buf)) {
                return buf;
            }
            return newDirectBuffer(buf);
        }
        throw new UnsupportedOperationException( "只接受ByteBuf类型参数");
    }

    @Override
    public boolean isActive() {
        DatagramChannel dc = javaChannel();
        return dc.isOpen() && dc.socket().isBound();
    }

    /**
     * Checks if the specified buffer is a direct buffer and is composed of a single NIO buffer.
     * (We check this because otherwise we need to make it a non-composite buffer.)
     */
    private static boolean isSingleDirectBuffer(ByteBuf buf) {
        return buf.isDirect() && buf.nioBufferCount() == 1;
    }


    @Override
    protected void doBind(SocketAddress localAddress) throws Exception { doBind0(localAddress); }
    private void doBind0(SocketAddress localAddress) throws Exception {
        if (PlatformDependent.javaVersion() >= 7) {
            SocketUtils.bind(javaChannel(), localAddress);
        } else {
            javaChannel().socket().bind(localAddress);
        }
    }


    @Override
    protected void doDisconnect() throws Exception { doClose(); }
    @Override
    protected void doClose() throws Exception {
        javaChannel().close();
        if (!kcpChannel.closeAnother) {
            kcpChannel.closeAnother = true;
            kcpChannel.unsafe().close(kcpChannel.unsafe().voidPromise());
        }
    }


    @Override
    public ChannelMetadata metadata() { return METADATA; }
    @Override
    protected void doFinishConnect() throws Exception { throw new Error(); }
    @Override
    protected DatagramChannel javaChannel() { return (DatagramChannel) super.javaChannel(); }
    @Override
    protected SocketAddress localAddress0() { return javaChannel().socket().getLocalSocketAddress(); }
    @Override
    protected SocketAddress remoteAddress0() {  return javaChannel().socket().getRemoteSocketAddress(); }
    @Override
    public ChannelConfig config() { return kcpChannel.config(); }
    @Override
    protected boolean continueOnWriteError() { return true; }
    @Override
    protected KcpClientUdpUnsafe newUnsafe() {
        return new KcpClientUdpUnsafe();
    }


    private class KcpClientUdpUnsafe extends AbstractNioUnsafe {
        private final List<Object> readBuf = new ArrayList<Object>();
        /**
         *  1）从udp口读取数据，读到readBuf列表中<br/>
         *  2）触发input（input中会把数据读到rcvbuf）<br/>
         *  3）触发receive（receive中会根据rcvWnd，把数据从rcvQue读给应用层。再根据都走了多少，从rcvBuf中补多少数据金rcvQue）<br/>
         *  4）用从rcvQue中读出来的数据，触发应用层的fireChannelRead()
         */
        @SuppressWarnings("deprecation")
		@Override
        public void read() {
            Assert.assertTrue(eventLoop().inEventLoop(), "Unsafe只能跑在EventLoop线程中");
            ChannelConfig config = config();
            ChannelPipeline pipeline = kcpChannel.pipeline();
            RecvByteBufAllocator.Handle allocatorHandle = recvBufAllocHandle();
            allocatorHandle.reset(config);

            boolean closed = false;
            //  读udp口数据产生的异常
            Throwable ex = null;
            try {
                //  把数据读取到readBuf中
                try {
                    do {
                        int r = doReadMessages(readBuf);
                        if (r == 0) {break;}
                        if (r < 0) {closed = true; break;}
                        allocatorHandle.incMessagesRead(r);
                    } while (allocatorHandle.continueReading());
                } catch (Exception e) {
                    ex = e;
                }
                //  把readBuf中的数据读到rcvBuf中，并触发一次input，再出发一次receive，再触发一次fireChannelRead
                //  input, receive, fireChannelRead 过程中触发的异常
                Throwable ex1 = null;
                int len = readBuf.size();
                try {
                    for (int i = 0 ; i < len ; i++) {
                    	//	把接收到的数据包还原成KcpSegment，保存进rcvBuf。再move进rcvQue
                    	//	要求每个数据包都至少是一份完整的kcp包（含头和体）
                        ByteBuf buf = (ByteBuf) readBuf.get(i);
                        kcpChannel.kcpInput(buf);
                    }
                    //  如果确实读到了数据，设置update时间戳
                    if (len > 0) { kcpChannel.kcpTsUpdate(TimestampTools.milliSeconds32()); }
                } catch (Throwable e) {
                    ex1 = e;
                }
                //  执行recv，把rcvQue中的数据读给应用层
                if (ex1 == null && kcpChannel.isActive()) {
                    boolean mergeSegmentBuf = RpcKcpConfigInfo.INSTANCE.isIkcpMergeSegment();
                    CodecOutputList<ByteBuf> recvBufList = null;
                    //  如果合并，用一个ByteBuf复制接收到的数据包，每读到一次触发一次fireChannelRead
                    try {
                        if (mergeSegmentBuf) {
                            ByteBufAllocator allocator = config.getAllocator();
                            int peekSize = 0;
                            while ((peekSize = kcpChannel.kcpPeekSize()) > 0) {
                                ByteBuf buf = allocator.ioBuffer(peekSize);
                                kcpChannel.kcpReceive(buf);
                                pipeline.fireChannelRead(buf);
                            }
                            pipeline.fireChannelReadComplete();
                        }
                        //  否则，把每个数据包add进recvBufList中，分别触发fireChannelRead。解码处自行处理拆包
                        else {
                            while (kcpChannel.kcpCanRecv()) {
                                if (recvBufList == null) {recvBufList = CodecOutputList.newInstance();}
                                kcpChannel.kcpReceive(recvBufList);
                            }
                            if (recvBufList != null) {
                            	NettyTools.fireChannelRead(kcpChannel, recvBufList);
                                recvBufList.recycle();
                            }
                        }
                    } catch (Throwable e) {
                        ex1 = e;
                    }
                }
                //  清理工作
                clearAndReleaseReadBuf();
                allocatorHandle.readComplete();
                //  处理异常
                if (ex != null) {
                    closeOnReadError(ex);
                    pipeline.fireExceptionCaught(ex);
                }
                if (ex1 != null) {
                    closed = true;
                    pipeline.fireExceptionCaught(ex1);
                }
                if (closed) {
                    inputShutdown = true;
                    if (isOpen()) {
                        close(voidPromise());
                    }
                }
            } finally {
                if (!config.isAutoRead()) {
                    clearAndReleaseReadBuf();
                }
            }
        }
        private void clearAndReleaseReadBuf() {
            int size = readBuf.size();
            for (int i = 0; i < size; i++) {
                Object msg = readBuf.get(i);
                ReferenceCountUtil.release(msg);
            }
            readBuf.clear();
        }

    }
}
